package com.chukun.flink.table.api.sql;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.concurrent.TimeUnit;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author chukun
 * @version 1.0.0
 * @description sql 使用处理时间窗口操作
 * @createTime 2022年06月03日 00:13:00
 */
public class GroupWindowProcessTimeTemplate {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 定义返回的数据列名称
        String[] fieldNames = new String[] {"id","number"};

        // 定义返回的数据列类型
        TypeInformation[] types = new TypeInformation[] {BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO};

        // 加载数据源
        DataStream<Row> dataStream = env.addSource(new SourceFunction<Row>() {
            @Override
            public void run(SourceContext<Row> ctx) throws Exception {
                int counter = 1;
                while (true) {
                    Row row = Row.of(counter % 2, counter);
                    ctx.collect(row);
                    System.out.println("send data : " + row);
                    TimeUnit.MICROSECONDS.sleep(3000);
                    counter++;
                }
            }

            @Override
            public void cancel() {}
        }).returns(Types.ROW(fieldNames, types));


        // 加载为table
        Table inputTable = tableEnv.fromDataStream(dataStream,$("id"), $("number"),$("autoAddTime").proctime());

        // 注册数据表
        tableEnv.createTemporaryView("test_proc_table", inputTable);

        // 使用窗口函数
        String sql = "select id, sum(number), " +
                "tumble_start(autoAddTime, interval '14' second) as wStart, " + // 窗口包含下界的开始时间
                "tumble_proctime(autoAddTime, interval '14' second) as wProcEnd," + // 窗口包含上界的结束时间
                "tumble_end(autoAddTime, interval '14' second) as wEnd " +  // 窗口不包含上界的结束时间
                "from test_proc_table " +
                "group by tumble(autoAddTime,interval '14' second), id";

        Table table = tableEnv.sqlQuery(sql);

        DataStream<Row> rowDataStream = tableEnv.toChangelogStream(table);

        rowDataStream.print("窗口时间分组");

        env.execute("GroupWindowProcessTimeTemplate");

    }
}
